package com.cn.thread.disruptor.multi;

import com.cn.thread.common.Consumer;
import com.cn.thread.disruptor.domain.Trade;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestMulti {

    public static void main(String[] args) throws InterruptedException {

        //1 创建RingBuffer
        RingBuffer<Trade> ringBuffer =
                RingBuffer.create(ProducerType.MULTI,
                        new EventFactory<Trade>() {
                            public Trade newInstance() {
                                return new Trade();
                            }
                        },
                        1024*1024,
                        new YieldingWaitStrategy());

        //2 通过ringBuffer 创建一个屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //3 创建多个消费者数组:
        ConsumerMulti[] consumers = new ConsumerMulti[10];
        for(int i = 0; i < consumers.length; i++) {
            consumers[i] = new ConsumerMulti("C" + i);
        }

        //4 构建多消费者工作池
        WorkerPool<Trade> workerPool = new WorkerPool<Trade>(
                ringBuffer,
                sequenceBarrier,
                new EventExceptionHandler(),
                consumers);

        //5 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        //6 启动workerPool
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        workerPool
                .start(threadPool);

        final CountDownLatch latch = new CountDownLatch(1);

        for(int i = 0; i < 100; i++) {
            final ProduceMulti producer = new ProduceMulti(ringBuffer);
            new Thread(new Runnable() {
                public void run() {
                    try {
                        latch.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    for(int j = 0; j<100; j++) {
                        producer.sendData(UUID.randomUUID().toString());
                    }
                }
            }).start();
        }

        Thread.sleep(2000);
        System.err.println("----------线程创建完毕，开始生产数据----------");
        latch.countDown();

        Thread.sleep(10000);

        System.err.println("任务总数:" + consumers[2].getCount());
        threadPool.shutdown();

    }

    static class EventExceptionHandler implements ExceptionHandler<Trade> {
        public void handleEventException(Throwable ex, long sequence, Trade event) {
        }

        public void handleOnStartException(Throwable ex) {
        }

        public void handleOnShutdownException(Throwable ex) {
        }
    }
}
